package com.wsx.ones.monitor.core.netty;

import com.wsx.ones.monitor.core.exec.PushCallback;
import com.wsx.ones.monitor.core.model.RpcRequest;
import com.wsx.ones.monitor.core.model.RpcResponse;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.concurrent.ConcurrentHashMap;

/**
 * netty客户端发送消息的核心句柄
 *
 * Created by wangshuaixin on 17/2/6.
 */
public class PushMessageHandler extends ChannelInboundHandlerAdapter {

    private ConcurrentHashMap<String, PushCallback> callbacks = new ConcurrentHashMap<String, PushCallback>();

    //执行的数据通道，需要由通道进行数据的传输
    private volatile Channel channel;


    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
        this.channel = ctx.channel();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
    }

    /**
     * 服务器端返回的数据结果
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        RpcResponse rpcResponse = (RpcResponse) msg;
        String requestId = rpcResponse.getRequestId();
        if (callbacks.containsKey(requestId)) {
            PushCallback callback = callbacks.get(requestId);
            //将结果放置在客户端的回调函数中
            callback.doEnd(rpcResponse);
            callbacks.remove(requestId);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        //super.exceptionCaught(ctx, cause);
        ctx.close();
    }

    /**
     * 客户端执行数据发送的真是操作，可以通过netty来实现
     * @param rpcRequest
     * @return
     */
    public PushCallback pushMessage(RpcRequest rpcRequest) {
        PushCallback callback = new PushCallback(rpcRequest);
        callbacks.put(rpcRequest.getRequestId(), callback);
        //数据写入通道并刷新到服务端
        channel.writeAndFlush(rpcRequest);
        return callback;
    }

    /**
     * 客户端关闭通道的处理
     * 仅仅在客户端程序停止时调用
     */
    public void close() {
        channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
    }

}
